package net.bwie.douyin.dwd.log1.Function;



import net.bwie.douyin.dwd.log1.bean.RoomGiftMoney;
import net.bwie.douyin.dwd.log1.bean.TikTokBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @ClassName：realTimeGiftFunction
 * @Author: Vvn
 * @Date: 2025/5/27 14:29
 * @Description: 必须描述类做什么事情, 实现什么功能
 */

public class realTimeGiftFunction implements
        WindowFunction<TikTokBean, RoomGiftMoney, String, TimeWindow> {

    private FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void apply(String s,
                      TimeWindow window,
                      Iterable<TikTokBean> input,
                      Collector<RoomGiftMoney> out) throws Exception {


        String windowStartTime = fastDateFormat.format(window.getStart());
        String windowEndTime = fastDateFormat.format(window.getEnd());

        Double giftMoney = 0.0;

        for (TikTokBean bean : input) {

            if ("gift".equals(bean.getEventType())) {
                giftMoney += bean.getGiftValue();
            }
        }

        out.collect(new RoomGiftMoney(s, windowStartTime, windowEndTime, giftMoney, System.currentTimeMillis()));
    }
}
